use async_stream::stream;
use async_trait::async_trait;

use crate::socket_capnp::{map, peer_http_request, peer_http_response};
use anyhow::anyhow;
use capnp::message::{ReaderOptions, TypedBuilder, TypedReader};
use futures::{SinkExt, StreamExt};
use rsocket_rust::{
    prelude::{Flux, Payload, RSocket, RSocketFactory},
    runtime, Result,
};
use rsocket_rust_transport_tcp::TcpClientTransport;
use std::sync::Arc;
use std::time::Duration;
use std::vec;
use tokio::sync::mpsc;
use tokio::{sync::Mutex, time::timeout};

use reqwest::header::{HeaderName, HeaderValue};
use serde_json::{json, Value};

use std::str::FromStr;

pub const DEFAULT_MIME_TYPE: &str = "application/binary";

pub async fn devices_device_get_peer_id() -> anyhow::Result<(String, String)> {
    //获取到本机的peer_id
    let peer_config = crate::config::get_peer_config().await?;
    //

    Ok((peer_config.peer_id, String::new()))
}

pub async fn devices_device_login(token: &String) -> anyhow::Result<String> {
    let (peer_id, _) = devices_device_get_peer_id().await?;
    //判断原来的链接是否还有效
    let has = {
        let cache = super::SOCKET_HANDLE.read().await;
        cache.is_some()
    };

    tracing::debug!("查看连接是否已经建立!");
    if has {
        tracing::debug!("连接已经建立!");
        //
        let arc_client = super::SOCKET_HANDLE
            .read()
            .await
            .as_ref()
            .map_or(Err(anyhow::anyhow!("")), |socket| Ok(socket))?
            .clone();

        let client = arc_client.lock().await;
        //测试连接

        let req = Payload::builder().build();
        tracing::debug!("开始请求:{:?}", req);
        if let Ok(response_result) =
            timeout(Duration::from_millis(30000), client.request_response(req)).await
        {
            let _response = response_result.map_err(|err| anyhow::anyhow!("{}", err))?;
            tracing::debug!("返回数据成功!");
        } else {
            tracing::debug!("请求超时!");
            //重新建立连接
            _create_socket(token, &peer_id).await?;
        }
    } else {
        tracing::debug!("连接未建立!");
        _create_socket(token, &peer_id).await?;
    }

    Ok(String::new())
}

async fn _create_socket(token: &String, peer_id: &String) -> anyhow::Result<()> {
    //创建rsocket client进行连接
    let token_clone = token.clone();

    let client = RSocketFactory::connect()
        .transport(TcpClientTransport::from(
            crate::constants::RRAI_CLOUD_SOCKET_HOST,
        ))
        .acceptor(Box::new(move || {
            // Return a responder.
            Box::new(RraiServerRSocket::new(&token_clone))
        }))
        .setup(
            Payload::builder()
                .set_metadata(peer_id.as_str())
                .set_data(
                    serde_json::json!({
                        "token": token,
                        "peer_id": peer_id,
                    })
                    .to_string(),
                )
                .build(),
        )
        .start()
        .await
        .map_err(|err| anyhow::anyhow!("Socket连接失败:{}", err))?;

    tracing::debug!("===============!");

    let mut cache = super::SOCKET_HANDLE.write().await;
    tracing::debug!("+++++++++!");

    match cache.as_mut() {
        Some(v) => *v = Arc::new(Mutex::new(client)),
        None => *cache = Some(Arc::new(Mutex::new(client))),
    }
    tracing::debug!("连接创建!");
    Ok(())
}

pub async fn devices_device_search(token: &String) -> anyhow::Result<String> {
    let url = format!("/devices/device/search");
    //请求的URL
    tracing::debug!("请求的URL:{}", url);
    let request_obj = json!({}).to_string();
    tracing::debug!("request:{:?}", request_obj);

    let res = crate::request::rrai_cloud_post(&url, token, request_obj).await?;
    Ok(res)
}

pub struct RraiServerRSocket {
    token: String,
}

impl RraiServerRSocket {
    pub fn new(token: &String) -> Self {
        Self {
            token: token.clone(),
        }
    }
}

#[async_trait]
impl RSocket for RraiServerRSocket {
    async fn metadata_push(&self, req: Payload) -> Result<()> {
        tracing::info!("{:?}", req);
        Ok(())
    }

    async fn fire_and_forget(&self, req: Payload) -> Result<()> {
        tracing::info!("{:?}", req);
        Ok(())
    }

    async fn request_response(&self, req: Payload) -> Result<Option<Payload>> {
        //{action:"",from:"",data:}
        let data_option = req.data();
        let payload = match data_option {
            Some(data) => {
                // //收到之后进行HTTP请求
                let token = self.token.clone();
                //request 转化
                let result = http_request(&token, data).await;
                // tracing::info!("{:?}", req.data_utf8());
                match result {
                    Ok(v) => {
                        tracing::debug!("返回的数据长度:{}", v.len());
                        Payload::builder().set_data(v).build()
                    }
                    Err(err) => Payload::builder()
                        .set_data_utf8(format!("{}", err).as_str())
                        .build(),
                }
            }
            None => Payload::builder().build(),
        };
        Ok(Some(payload))
    }

    fn request_stream(&self, req: Payload) -> Flux<Result<Payload>> {
        tracing::info!("{:?}", req);
        Box::pin(stream! {
            yield Ok(req);
        })
    }

    fn request_channel(&self, mut reqs: Flux<Result<Payload>>) -> Flux<Result<Payload>> {
        let (sender, mut receiver) = mpsc::unbounded_channel();
        runtime::spawn(async move {
            while let Some(it) = reqs.next().await {
                tracing::info!("runtime:{:?}", it);
                sender.send(it).unwrap();
            }
        });
        Box::pin(stream! {
            while let Some(it) = receiver.recv().await {
                tracing::info!("runtime2:{:?}", it);
                yield it;
            }
        })
        // or returns directly
        // reqs
    }
}

// request: peer_http_request::Reader<'_>,
pub async fn http_request(_token: &String, data: &bytes::Bytes) -> Result<Vec<u8>> {
    let (uri, version, method, headers, body) = {
        let reader =
            capnp::serialize_packed::read_message(data.to_vec().as_slice(), ReaderOptions::new())?;

        let message_reader = TypedReader::<_, peer_http_request::Owned>::new(reader);
        let rr = message_reader.get()?;

        let head = rr.reborrow().get_head()?;
        //uri
        let uri = head.reborrow().get_uri()?.to_string()?;

        //headers
        let mut headers = reqwest::header::HeaderMap::new();

        head.get_headers()?.get_entries()?.iter().for_each(|entry| {
            let key = entry.get_key();
            let value = entry.get_value();

            if key.is_ok() && value.is_ok() {
                headers.insert(
                    HeaderName::from_str(key.unwrap().to_str().unwrap()).unwrap(),
                    HeaderValue::from_str(value.unwrap().to_str().unwrap()).unwrap(),
                );
            }
        });

        //version
        let request_version = head.reborrow().get_version()?.to_str()?.to_uppercase();
        let version = if request_version == "HTTP/0.9" {
            http::Version::HTTP_09
        } else if request_version == "HTTP/1.0" {
            http::Version::HTTP_10
        } else if request_version == "HTTP/1.1" {
            http::Version::HTTP_11
        } else if request_version == "HTTP/2.0" {
            http::Version::HTTP_2
        } else if request_version == "HTTP/3.0" {
            http::Version::HTTP_3
        } else {
            http::Version::HTTP_11
        };

        //method
        let request_method = head.reborrow().get_method()?.to_str()?.to_uppercase();
        let method = reqwest::Method::from_str(request_method.as_str())?;

        //body
        let request_body = rr.reborrow().get_body()?.to_vec();
        let body = bytes::Bytes::from(request_body);
        //
        let uri = if uri.ends_with("/ws") { uri } else { uri };
        tracing::debug!("uri:{:?}", uri);
        //method
        (uri, version, method, headers, body)
    };
    //收到之后进行HTTP请求
    let response = reqwest::Client::new()
        .request(method, uri)
        .version(version)
        .headers(headers)
        .body(body)
        .send()
        .await
        .map_err(|err| anyhow!("请求失败:{}", err))?;

    let response_headers = response.headers().clone();
    let response_status = response.status().as_u16();

    let content = response
        .bytes()
        .await
        .map_err(|err| anyhow!("请求失败:{}", err))?
        .to_vec()
        .clone();

    //
    let mut message = ::capnp::message::Builder::new_default();

    let mut response_builder = TypedBuilder::<peer_http_response::Owned>::new_default();
    let mut response_root = response_builder.get_root()?;
    //status
    response_root.set_status(response_status);

    //headers
    let mut headers_builder =
        TypedBuilder::<map::Owned<capnp::text::Owned, capnp::text::Owned>>::new_default();

    let mut headers_root = headers_builder.get_root()?;

    let headers = message.init_root::<map::Builder<capnp::text::Owned, capnp::text::Owned>>();
    let mut entries = headers.init_entries(response_headers.len().try_into()?);

    let mut temp = 0;

    response_headers
        .iter()
        .for_each(|(header_key, header_value)| {
            let mut entry = entries.reborrow().get(temp);
            if let Ok(value) = header_value.to_str() {
                let _ = entry.set_key(capnp::text::Reader(header_key.as_str().as_bytes()));
                let _ = entry.set_value(capnp::text::Reader(value.as_bytes()));
            }
            temp = temp + 1;
        });
    headers_root.set_entries(entries.into_reader())?;

    response_root.set_headers(headers_builder.get_root_as_reader()?)?;
    //body
    response_root.set_body(content.as_slice());

    let mut buffer = vec![];
    capnp::serialize_packed::write_message(&mut buffer, response_builder.borrow_inner())?;

    Ok(buffer)
}
